package com.lbb.rocketmq;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

public class Customer {
	
	
	Logger logger = LoggerFactory.getLogger(getClass());
	
	
	
	public void run(){
		
		
		//这里参数是自定义的  不要与生产者一样  
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        logger.debug("消费者： {} 开始接收数据...");
        try {
            // 设置topic和标签
            consumer.subscribe("MessagePlantform", "TagA");
//            consumer.setVipChannelEnabled(false);
            
            /**
             * CONSUME_FROM_LAST_OFFSET //默认策略，从该队列最尾开始消费，即跳过历史消息
               CONSUME_FROM_FIRST_OFFSET //从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
               CONSUME_FROM_TIMESTAMP//从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            
            //监听消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                        ConsumeConcurrentlyContext Context) {
                	
                	for (MessageExt msg :list){
                		logger.debug("{}收到数据id{}:{},topic:{}",Thread.currentThread().getName() ,
                				msg.getMsgId(), new String(msg.getBody()),msg.getTopic());
                	}
                	
                	
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
		
	}
	
	public static void main(String[] args) {
		new Customer().run();

	}
//
}
